//package com.java.springboot.elasticsearch.canal;
//
//import com.alibaba.otter.canal.client.CanalConnector;
//import com.alibaba.otter.canal.client.CanalConnectors;
//import com.google.common.collect.Lists;
//import org.springframework.beans.factory.DisposableBean;
//import org.springframework.context.annotation.Bean;
//import org.springframework.stereotype.Component;
//
//import java.net.InetSocketAddress;
//
///**
// * @author zhangtengfei
// * @date 2021/10/13 11:45
// */
//@Component
//public class CanalClient implements DisposableBean {
//
//    private CanalConnector canalConnector;
//
//    @Bean
//    public CanalConnector getCanalConnector() {
//        // 即使我们用的是集群，其实同时也只有一个canal能工作，只有当他挂了 才会启用其他的，所以就是一个备份
//        canalConnector = CanalConnectors.newClusterConnector(
//                Lists.newArrayList(new InetSocketAddress("127.0.0.1", 11111)),
//                "example",
//                "root",
//                "123456"
//        );
//        canalConnector.connect();
//        /**
//         * 指定要监听的表。格式{database}.{table}。这里 *.* 全部数据库的全部表
//         * 如果要监听的表 很多，建议用多线程，每一个canal服务只处理几张表就行
//         * canalConnector.subscribe("elasticsearch.read_book_pd");
//         */
//        canalConnector.subscribe("*.*");
//        // 回滚到未进行ack的地方, 下次fetch的时候, 可以从最后一个没有ack的地方开始拿
//        canalConnector.rollback();
//        return canalConnector;
//    }
//
//    @Override
//    public void destroy() throws Exception {
//        if (canalConnector != null) {
//            canalConnector.disconnect();
//        }
//    }
//}
//
